package com.xiaojie.hadoop.mapreduce.flow;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author 熟透的蜗牛
 * @version 1.0
 * @description: 流量mapper
 * @date 2024/12/27 10:32
 */
public class FlowBeanMapper extends Mapper<LongWritable, Text, Text, FlowBean> {

    //定义一个输出的key
    private Text outKey = new Text();
    //定义输出的value 即 FlowBean
    private FlowBean outValue = new FlowBean();

    /**
     * @param key     map的输入值偏移量
     * @param value   map 的输入value
     * @param context
     * @description:
     * @return: void
     * @author 熟透的蜗牛
     * @date: 2024/12/27 10:35
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {

        //获取一行数据
        String line = value.toString();
        //切割数据
        String[] split = line.split("\t");

        //抓取我们需要的数据:手机号,上行流量,下行流量
        String phone = split[1];  //手机号
        //上行流量 ，由于有的数据没有，这里从后面取值
        Long upFlow = Long.parseLong(split[split.length - 3]);
        Long downFlow = Long.parseLong(split[split.length - 2]);

        //封装输出结果
        //设置输出的key
        outKey.set(phone);
        //设置输出的value
        outValue.setUpFlow(upFlow);
        outValue.setDownFlow(downFlow);
        outValue.setSumFlow();
        //写出outK outV
        context.write(outKey, outValue);
    }
}